package com.sys.kafka.factory;

import com.sys.kafka.core.producer.IProducerHandler;
import com.sys.kafka.core.producer.ProducerHandler;
import com.sys.midware.logger.Logger;
import com.sys.midware.logger.LoggerFactory;

/**
 * 获取默认的消息生产实力 <b>ClassName：</b> ProducerFactory <br/>
 * <b>Description：</b> TODO <br/>
 * <b>@author：</b> mobing <br/>
 * <b>@date：</b> 2015年10月19日 下午2:54:26 <br/>
 * <b>@version: </b> <br/>
 */
public class ProducerFactory {
    private static final Logger logger = LoggerFactory.getLogger(ProducerFactory.class);

    /**
     * IProducer 接口定义
     */
    private static IProducerHandler producerHandler = null;

    /**
     * 获取默认消息生产实例
     * @return IProducer
     */
    public static IProducerHandler getProducerInstace() {
        return getProducerInstace(null);
    }

    /**
     * 获取消息生产实例
     * @param producerGroup
     *            消息发送的组名
     * @return IProducer
     */
    public static IProducerHandler getProducerInstace(String producerGroup) {
      
        return getProducerInstace(null,producerGroup);
    }
    
    /**
     * 获取消息生产实例
     * @param producerGroup
     *            消息发送的组名
     * @return IProducer
     */
    public static IProducerHandler getProducerInstace(String clientId,String producerGroup) {
        if (null != producerHandler) {
            return producerHandler;
        }
        
        synchronized (ProducerFactory.class) {
            if (null == producerHandler) {
                producerHandler = new ProducerHandler();
                
                Runtime.getRuntime().addShutdownHook(new CleanHook());
            }
        }
        return producerHandler;
    }
    
    static void clean() {
        try {
//            producerHandler.shutdown();
        } catch (Exception e) {
            logger.error("clean producer error!", e);
        }
    }

    static class CleanHook extends Thread {
        public void run() {
            clean();
        }
    }
}
